package ntu.p2p.past;

import java.io.IOException;

import ntu.p2p.past.messaging.UpdateObjectMessage;
import rice.Continuation;
import rice.Continuation.StandardContinuation;
import rice.environment.logging.Logger;
import rice.p2p.commonapi.Id;
import rice.p2p.commonapi.Message;
import rice.p2p.commonapi.Node;
import rice.p2p.commonapi.NodeHandle;
import rice.p2p.commonapi.NodeHandleSet;
import rice.p2p.commonapi.rawserialization.InputBuffer;
import rice.p2p.past.PastContent;
import rice.p2p.past.PastContentHandle;
import rice.p2p.past.PastException;
import rice.p2p.past.PastPolicy;
import rice.p2p.past.PastPolicy.DefaultPastPolicy;
import rice.p2p.past.rawserialization.SocketStrategy;
import rice.persistence.StorageManager;

@SuppressWarnings("unchecked")
public class PastImpl extends rice.p2p.past.PastImpl implements Past {
	
	protected class PastDeserializer extends rice.p2p.past.PastImpl.PastDeserializer {
		public Message deserialize(InputBuffer buf, short type, int priority,
    		NodeHandle sender) throws IOException {
      if (type == 8) {
      	System.out.println(buf.toString());
      	return UpdateObjectMessage.build(buf, endpoint);
      } else {
      	return super.deserialize(buf, type, priority, sender);
      }
    }   
	}
	
	protected UpdateApplicator updateApplicator;
	
	/**
	 * Constructor for Past, using the default policy
	 * 
	 * @param node
	 *            The node below this Past implementation
	 * @param manager
	 *            The storage manager to be used by Past
	 * @param replicas
	 *            The number of object replicas
	 * @param instance
	 *            The unique instance name of this Past
	 */
	public PastImpl(Node node, StorageManager manager, int replicas,
			String instance, UpdateApplicator updateApplicator) {
		super(node, manager, replicas, instance, new DefaultPastPolicy());
		this.updateApplicator = updateApplicator;
		this.endpoint.setDeserializer(new PastDeserializer());
	}

	/**
	 * Constructor for Past
	 * 
	 * @param node
	 *            The node below this Past implementation
	 * @param manager
	 *            The storage manager to be used by Past
	 * @param replicas
	 *            The number of object replicas
	 * @param instance
	 *            The unique instance name of this Past
	 */
	public PastImpl(Node node, StorageManager manager, int replicas,
			String instance, PastPolicy policy, UpdateApplicator updateApplicator) {
		super(node, manager, null, replicas, instance, policy, null);
		this.updateApplicator = updateApplicator;
		this.endpoint.setDeserializer(new PastDeserializer());
	}

	@Override
	public void deliver(Id id, Message message) {
		if (message instanceof UpdateObjectMessage) {
			System.out.println("updating...........");
			final UpdateObjectMessage updateObject = (UpdateObjectMessage) message;
			final UpdateApplicator applicator = this.updateApplicator;
			this.lookupHandle(updateObject.getContentId(), this.getLocalNodeHandle(),
					new Continuation<PastContentHandle, Exception>() {

				@Override
				public void receiveException(Exception exception) {
					// TODO Auto-generated method stub

				}

				@Override
				public void receiveResult(PastContentHandle result) {
					inserts++;
					final Id objectId = result.getId();
					lockManager.lock(objectId, new StandardContinuation(this) {
						@Override
						public void receiveResult(Object result) {
							storage.getObject(objectId, new StandardContinuation(parent) {

								@Override
								public void receiveResult(Object result) {
									PastContent content = (PastContent) result;
									final PastContent updatedContent = applicator
									.applyUpdate(content, updateObject.getDifference());
									storage.store(objectId, null, updatedContent,
											new StandardContinuation(parent) {
										public void receiveResult(Object result) {
											lockManager.unlock(objectId);
										}
									});
								}
							});
						}
					});
				}
			});
		} else {
			super.deliver(id, message);
		}
	}

	@Override
	public void update(final Id id, final UpdateObjectMessage update, final Continuation command) {
		getHandles(id, replicationFactor+1, new StandardContinuation(command) {
			@Override
			public void receiveResult(Object result) {
				// TODO Auto-generated method stub
		        NodeHandleSet replicas = (NodeHandleSet) result;
		        if (logger.level <= Logger.FINER) logger.log("Received replicas " + replicas + " for id " + id);
		        
		        // then we send inserts to each replica and wait for at least
		        // threshold * num to return successfully
		        MultiContinuation multi = new MultiContinuation(parent, replicas.size()) {
		          public boolean isDone() throws Exception {
		            int numSuccess = 0;
		            for (int i=0; i<haveResult.length; i++) 
		              if ((haveResult[i]) && (Boolean.TRUE.equals(result[i])))  
		                numSuccess++;
		            
		            if (numSuccess >= (SUCCESSFUL_INSERT_THRESHOLD * haveResult.length)) 
		              return true;
		            
		            if (super.isDone()) {
		              for (int i=0; i<result.length; i++) 
		                if (result[i] instanceof Exception)
		                  if (logger.level <= Logger.WARNING) logger.logException("result["+i+"]:",(Exception)result[i]);
		              
		              throw new PastException("Had only " +  numSuccess + " successful inserts out of " + result.length + " - aborting.");
		            }
		            return false;
		          }
		          
		          public Object getResult() {
		            Boolean[] b = new Boolean[result.length];
		            for (int i=0; i<b.length; i++)
		              b[i] = new Boolean((result[i] == null) || Boolean.TRUE.equals(result[i]));
		            
		            return b;
		          }
		        };
		        
		        for (int i=0; i<replicas.size(); i++) {
		          NodeHandle handle = replicas.getHandle(i);
		          System.out.println("send update message..");
		          Continuation c = new NamedContinuation("Send Update Message to " + replicas.getHandle(i) + " for " + id, multi.getSubContinuation(i));
		          endpoint.route(null, update, handle);
		        }
		        command.receiveResult(null);
			}
		
		});
	}
}
